package com.atguigu.gmall.realtime.bootstrap

import java.util

import com.alibaba.fastjson.JSONObject
import com.atguigu.gmall.realtime.util.{MyKafkaSender, MySQLUtil, RedisUtil}
import redis.clients.jedis.Jedis

object GmallDimBootstrap {


  //0 获取维度表清单
  //1 读取对应的mysql表 提取维度数据
  //2 仿照canal的格式 转换格式
  //3  发送kafka

  def main(args: Array[String]): Unit = {
    //0 获取维度表清单
    val jedis: Jedis = RedisUtil.getJedisClient
    val dimTablekey="DIM_TABLES"
    val dimTables: util.Set[String] = jedis.smembers(dimTablekey)
    //1 读取对应的mysql表 提取维度数据
    import scala.collection.JavaConverters._
    for (dimTableName <- dimTables.asScala) {
      val sql=s" select * from $dimTableName"
      val dimTableRows: util.List[JSONObject] = MySQLUtil.queryList(sql)

      //2 仿照canal的格式 转换格式
      val jsonForKafka=new JSONObject()
      jsonForKafka.put("data",dimTableRows)
      jsonForKafka.put("table",dimTableName)
      jsonForKafka.put("type","INSERT")
      //3  发送kafka
      MyKafkaSender.send("ODS_BASE_DB_C",jsonForKafka.toJSONString)

    }
    MyKafkaSender.close()
    jedis.close()

  }

}
